package com.zyh.alpacaSse.service;


import com.fasterxml.jackson.databind.ObjectMapper;
import com.zyh.alpacaSse.config.AlpacaSseProperties;
import com.zyh.alpacaSse.pojo.JournalEvent;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.ReactiveRedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.util.LinkedMultiValueMap;
import org.springframework.util.MultiValueMap;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.util.retry.Retry;
import reactor.util.retry.RetryBackoffSpec;
import org.springframework.web.reactive.function.client.WebClient;

import java.time.Duration;
import java.util.Collections;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.atomic.AtomicReference;

/**
 * @Classname AlpacaSseService
 * @Description TODO()
 * @Date 2025-09-03 22:50
 * @Created by 86177
 */
@Slf4j
@Service
@RequiredArgsConstructor
public class AlpacaSseService {
    private final WebClient webClient;
    private final AlpacaSseProperties properties;
    private final ObjectMapper objectMapper;
    private final ReactiveRedisTemplate<String, String> redisTemplate;
    // 提取 Redis key 前缀为常量
    private static final String EVENT_KEY_PREFIX = "event:";

    private final AtomicReference<String> lastEventId = new AtomicReference<>();

    public Flux<JournalEvent> streamEvents() {
        //使用Flux.defer()创建一个延迟执行的Flux，确保每次订阅时都会创建新的连接，而不是共享之前的连接
        return Flux.defer(() -> {
            // 1. 构建查询参数
//            Map<String, String> queryParams = Collections.emptyMap();

            // 2. 创建 SSE 请求
            return webClient.get()
                    .uri(uriBuilder -> uriBuilder
                            //设置起始时间，从该时间点开始接收事件
                            .queryParam("since", "2023-01-01T00:00:00Z")
                            //如果存在lastEventId，则添加since_id参数，用于断点续传
                            .queryParamIfPresent("since_id", Optional.ofNullable(lastEventId.get()))
                            .build())
                    //发起HTTP请求
                    .retrieve()
                    //将响应体转换为一个持续的数据流
                    .bodyToFlux(String.class)
                    //为监控添加名称和指标
                    .name("alpaca-sse")
                    // 处理背压，缓冲最多1000个事件，防止生产者过快导致消费者崩溃
                    .onBackpressureBuffer(1000)
                    //将每个事件字符串解析为JournalEvent对象
                    .flatMap(this::parseEvent)
                    //只处理类型为"journal/status"的事件
                    .filter(event -> "journal/status".equals(event.getType()))
                    //数据去重
                    .transform(this::deduplicateEvents)
                    // 成功地处理一个事件后，它就会把这条消息的ID：lastEventId记录下来，用于断点续传
                    .doOnNext(event -> lastEventId.set(event.getId()))
                    .doOnSubscribe(sub -> log.info("Starting SSE connection"))
                    .doOnCancel(() -> log.warn("SSE connection cancelled"))
                    .doOnTerminate(() -> log.error("SSE connection terminated"))
                    .doOnError(e -> log.error("SSE connection error", e))
                    //接失败时根据策略重试
                    .retryWhen(createRetryStrategy())
                    // 共享连接，允许多个订阅者
                    .share();
        });
    }

    /**
     * 解析事件数据为JournalEvent对象
     * @param eventData
     * @return
     */
    private Mono<JournalEvent> parseEvent(String eventData) {
        // 包装成一个异步任务，让它慢慢干活，干完给我一个结果
        return Mono.fromCallable(() -> objectMapper.readValue(eventData, JournalEvent.class))
                .onErrorResume(e -> {
                    log.error("Failed to parse event: {}", eventData, e);
                    // TODO 根据业务需求选择是否抛出异常或继续处理
                    // return Mono.error(new omsException("Failed to parse SSE event data: " + eventData, e));
                    // 当前选择是仍进去垃圾桶，丢弃
                    return Mono.empty();
                });
    }

    /**
     * 数据去重
     * @param flux 原材料流
     * @return  Flux<JournalEvent> 输出流
     */
    private Flux<JournalEvent> deduplicateEvents(Flux<JournalEvent> flux) {
        return flux.filterWhen(event ->
                redisTemplate.opsForValue().setIfAbsent(
                        // CAS 操作，如果key不存在则设置，并返回true，否则返回false
                        EVENT_KEY_PREFIX + event.getId(),
                        "1",
                        // 设置ttl时间，当前设置仅生存 1 hours
                        properties.getDedupe().getCacheTtl()
                ).map(Boolean::booleanValue) // 将结果映射为Mono<Boolean>
        );
    }

    // 实现指数退避重试机制，最大重试次数和延迟时间可配置
    private RetryBackoffSpec createRetryStrategy() {
        // 配置初始延迟（50ms）和最大重试次数(1000)
        return Retry.backoff(properties.getReconnect().getMaxAttempts(),
                        Duration.ofMillis(properties.getReconnect().getInitialDelay()))
                //设置最大退避时间，指数退避拉满60s后，每60s试一次，直到达到最大重试次数
                .maxBackoff(Duration.ofMillis(properties.getReconnect().getMaxDelay()))
                //多客户端请求可能引发重试风暴，因此添加 0.5 的抖动因子，即实际等待时间 = 等待时间 * (1 ± 抖动因子)
                .jitter(0.5)
                //每次重试前记录警告日志；
                .doBeforeRetry(retry -> log.warn("Retry attempt #{}/{} after failure",
                        // 当前第几次重试
                        retry.totalRetries() + 1,
                        // 最大重试次数
                        properties.getReconnect().getMaxAttempts(),
                        // 导致重试的异常原因
                        retry.failure()))
                //超过最大重试次数时抛出异常
                .onRetryExhaustedThrow((spec, signal) ->
                        new RuntimeException("Max retry attempts exceeded"));
    }
}
